package iot;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import iot.entity.Location;
import iot.jtt808.DataPack;
import iot.jtt808.platform.PlatCommonResp;
import iot.jtt808.platform.TerminalRegisterResp;
import iot.jtt808.terminal.LocationReport;
import netty.session.NettySession;
import netty.session.SessionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by dave on 18-7-20 下午4:26.
 */
public class MessageProcessor {
    private static final Logger L = LoggerFactory.getLogger(MessageProcessor.class);
    private SessionManager sessionManager;
    private DataDispatcher dispatcher;
    private AtomicLong locCounter = new AtomicLong(0);
    private AtomicLong regCounter = new AtomicLong(0);
    private AtomicLong counter = new AtomicLong(0);

    public MessageProcessor(SessionManager sessionManager, DataDispatcher dispatcher) {
        this.sessionManager = sessionManager;
        this.dispatcher = dispatcher;
    }

    @SuppressWarnings("unchecked")
    public void process(NettySession session, DataPack<?> dp) {
        switch (dp.getMsgId()) {
            case 0x0002: { // 终端心跳
                succeed(session.getChannel(), dp);
                break;
            }
            case 0x0100: { // 终端注册
                DataPack<TerminalRegisterResp> re = new DataPack<>();
                re.setMsgId(0x8100);
                re.setSim(dp.getSim());
                TerminalRegisterResp res = new TerminalRegisterResp();
                res.setTerminalSequenceNo(dp.getSequenceNo());
                res.setResult(0);
                res.setAuth(UUID.randomUUID().toString());
                re.setBody(res);
                session.getChannel().writeAndFlush(Unpooled.wrappedBuffer(re.encode()));
//                if (!dispatcher.dispatch(Registration.fromDataPack((DataPack<TerminalRegister>) dp))) {
//                    // TODO 队列满
//                    regCounter.incrementAndGet();
//                }
                break;
            }
            case 0x0200: { // 位置信息上报
                succeed(session.getChannel(), dp);
                if (!dispatcher.dispatch(Location.fromDataPack((DataPack<LocationReport>) dp))) {
                    // TODO 队列满
                    locCounter.incrementAndGet();
                }
                break;
            }
        }
        long c = counter.incrementAndGet();
        if (c % 20000 == 0) {
            L.info("已处理(万包)：" + (c/10000) + "。丢弃：" + regCounter.get() + ", " + locCounter.get());
        }
    }

    private void succeed(Channel channel, DataPack<?> dp) {
        DataPack<PlatCommonResp> re = new DataPack<>();
        re.setMsgId(0x8001);
        re.setSim(dp.getSim());
        NettySession s = sessionManager.getSession(dp.getSim());
        if (s != null) {
            re.setSequenceNo(s.nextSequenceNo());

            PlatCommonResp body = new PlatCommonResp();
            body.setTerminalSequenceNo(dp.getSequenceNo());
            body.setTerminalMsgId(dp.getMsgId());
            body.setResult(DataPack.success);
            re.setBody(body);
            channel.writeAndFlush(Unpooled.wrappedBuffer(re.encode()));
        }
    }
}
